import json
from package.connector.kafka_db import kafka_consumer
from package.connector.sql_db import db
from package.connector.redis_db import rdb
from setting import setting 
from package.fastapi.job import scheduler_job
from common.notify import NotifyMap


class SystemTask:
    """系统任务
    定时任务
    Kafka任务
    """
    pass


class CronTask(SystemTask):
    """定时任务"""

    @classmethod
    def load_cron(cls):
        """加载 数据库内定时任务"""
        from router_collector.source.view import SourceView
        with db.SessionLocal() as s:
            for item in SourceView.get_items(session=s, filter_=[SourceView.category.in_(SourceView.CategoryCollector)]).all():
                SourceView.add_task(item)


class KafkaTask(SystemTask):
    NotifyTopic = 'system_notify_task'

    @classmethod
    def execute(task):
        """执行任务"""
        notify_configure = task['notify_configure']
        message = task['message']
        
        notify = NotifyMap[notify_configure['type']](**notify_configure)
        notify.notify(**message)

    @classmethod
    def consume(cls):
        """
        消费数据 获取所有发现的数据源
        添加新增的数据发现
        """
        kafka_consumer.consumer.subscribe([cls.NotifyTopic])

        while True:
            messages = kafka_consumer.consume(num_messages=10, timeout=5)
            if not messages:
                continue

            for item in messages:
                if item.error():
                    continue
                
                task = json.loads(item.value())
                cls.execute(**task)


if __name__ == '__main__':
    db.init(url=setting.pg_uri, pool_pre_ping=True)
    rdb.init(host=setting.redis_host, password=setting.redis_password)
    kafka_consumer.init(**{'bootstrap.servers': setting.kafka_servers, **setting.kafka_options, 'group.id': 'system_task'})
    scheduler_job.init_config(store_configure=dict(host=setting.redis_host, password=setting.redis_password,
                                          jobs_key='collector2:apscheduler',
                                          run_times_key='collector2:apscheduler.runtime'),
                    default_configure={'coalesce': True, 'max_instances': 1}, is_background=True)
    
    CronTask.load_cron()
    scheduler_job.scheduler.start()
    KafkaTask.consume()
